package net.dgi.common.match.impl;

import net.dgi.common.ioc.Inject;
import net.dgi.common.ioc.InjectSource;
import net.dgi.common.ioc.Injector;
import net.dgi.common.match.MatchEngine;
import net.dgi.common.match.impl.*;

/**
 * Viva IntelliJ IDEA.
 * User: Deian Date: Sep 30, 2010 Time: 10:15:22 PM
 */
public class MatchEngineImpl implements MatchEngine {
    @Inject(InjectSource.SHARE)
    protected MatchContext context;

    @Inject(InjectSource.SHARE)
    protected FunctionRegistry functionRegistry;

    @Inject(InjectSource.SHARE)
    Thread matchExecutorThread = new Thread(new MatchExecutor());
    Thread matcherThread = new Thread(new Matcher());

    public MatchEngineImpl() {
        Injector.register(InjectSource.SHARE, FunctionRegistry.class, new FunctionRegistryImpl());
        Injector.register(InjectSource.SHARE, MatchContext.class, new MatchContext());
        Injector.inject(this);
        start();
    }

    private void start() {
        matcherThread.start();
        matchExecutorThread.start();
    }

    /**
     * Requests shutdown from all internal threads. If there are clients waiting for result they might receive InterruptedException
     */
    public void requesShutdown() {
        context.shutdown();
    }

    /**
     * NonBlocking! Drop data in the engine and it will process the data until exhaust all available functions.
     * the result can be extracted with poll.  Awakes the threads to process it
     *
     * @param object Object to process
     */
    public synchronized void add(Object object) {
        context.getIncomingQueue().add(new MatchRequest(object));
        notifyAll();
    }

    /**
     * Blocking method! drops the data into the engine and blocks until we get result
     *
     * @param object Object to processing queue and awakes the threads to process it
     * @return Object result of the processing
     * @throws Throwable
     */
    public synchronized Object request(Object object) throws Throwable {
        add(object);
        return poll();
    }

    /**
     * Add functionality to the engine and awakes all threads!
     *
     * @param function
     */
    public synchronized void addFunctions(Object function) {
        functionRegistry.register(function);
        notifyAll();
    }

    /**
     * Blocking method. Get response from the engine
     *
     * @return Object result of the processing
     * @throws Throwable
     */
    public synchronized Object poll() throws Throwable {
        for (; ;) {
            if (context.hasShutdownSignal()) {
                throw new InterruptedException();
            }
            MatchResult result = context.getOutgoingQueue().poll();
            try {
                if (result == null) {
                    wait();
                }
            } catch (InterruptedException e) {
                continue;
            }
            if (result.getStatus() != null) {
                throw result.getStatus();
            }
            return result.getMatchRequest().getData();
        }
    }

    public int size() {
        return context.getOutgoingQueue().size();
    }
}
